package com.stefanie.kafkademo.demo;

import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

/**
 * @author: stefanie-zy
 * @time: 2023/12/9-16:45
 * @description: kafka-生产者
 */
@Slf4j
public class MyProducer {
    private final static String TOPIC_NAME = "stefanie-zy-test"; // 主题

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        // 存放键值对属性
        Properties properties = new Properties();

        // 集群使用逗号隔开
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");

        // ack配置[数值必须是字符串不能是数字类型]
        properties.put(ProducerConfig.ACKS_CONFIG, "1");

        // 重试机制
        properties.put(ProducerConfig.RETRIES_CONFIG, 3); // 重试次数
        properties.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 300); // 重试间隔时间

        // 缓冲区配置32M[设置RecordAccumulator的大小，便于send可以批量发送]
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        // 线程拉取数据大小配置16K[ProducerBatch，设置批次的大小，可以通过BufferPool来管理]
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        // 如果线程拉取不到16384[16k]的数据，间隔10ms也会将拉取到的数据发送
        properties.put(ProducerConfig.LINGER_MS_CONFIG, 10);

        // 设置消息发送但是未收到响应的个数，如果超过5则不能继续向这个连接发送数据[InFlightRequest]
        properties.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);

        // 设置Key、Value的序列化
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 设置分区[自定义分区：MyPartition]
        properties.put(ProducerConfig.CONFIG_PROVIDERS_CONFIG, MyPartition.class.getName());

        // 设置自定义拦截器
        properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, MyProducerInterceptor.class.getName());

        // 设置发消息的客户端
        Producer<String, String> producer = new KafkaProducer<>(properties);

        // 发送消息
        int msgNum = 5;
        // 多线程结果统计
        final CountDownLatch countDownLatch = new CountDownLatch(msgNum);
        int i = 1;
        while (i <= msgNum) {
            Order order = new Order((long) i, i, "stefanie-" + i);
            // 创建生产者
            // 未指定发送分区，通过order.getId()来计算分区信息
            ProducerRecord<String, String> producerRecord = new ProducerRecord<>(TOPIC_NAME, order.getId().toString(), JSONUtil.toJsonStr(order));
            // 指定分区[0为指定分区]
            ProducerRecord<String, String> producerRecord1 = new ProducerRecord<>(TOPIC_NAME, 0, order.getId().toString(), JSONUtil.toJsonStr(order));

            // 发送消息——同步
            str(producer, producerRecord1);
            // 发送消息——异步
            asm(producer, producerRecord, countDownLatch);
            i++;
        }
        // 消息发送完成之后关闭客户端
        producer.close();
    }

    /**
     * 同步发送
     */
    private static void str(Producer<String, String> producer, ProducerRecord<String, String> producerRecord) throws ExecutionException, InterruptedException {
        // 发送数据并获取返回值
        RecordMetadata recordMetadata = producer.send(producerRecord).get();
        log.info("kakfa同步发送消息结果：topic:{}；partition:{}；offset{}", recordMetadata.topic(), recordMetadata.partition(), recordMetadata.offset());
    }

    /**
     * 异步发送
     */
    private static void asm(Producer<String, String> producer, ProducerRecord<String, String> producerRecord, CountDownLatch countDownLatch) throws InterruptedException {
        producer.send(producerRecord, (recordMetadata, e) -> {
            if (e != null) {
                log.error("kafka异步发送消息失败：{}", (Object) e.getStackTrace());
                return;
            }
            if (recordMetadata != null) {
                log.info("kakfa异步发送消息结果：topic:{}；partition:{}；offset{}", recordMetadata.topic(), recordMetadata.partition(), recordMetadata.offset());
            }
            // 每次发送成功-1
            countDownLatch.countDown();
        });

        // 等待异步响应，否则线程结束看不到上述日志打印信息
//        Thread.sleep(1000L);

        // 每次线程阻塞，当countDownLatch=0的时候不阻塞；每次等待时间为5s
        boolean await = countDownLatch.await(1, TimeUnit.SECONDS);
        log.info("countDownLatch={}；await={}", countDownLatch.getCount(), await);
        if (await) {
            // 确保消息发送完成之后关闭客户端
            producer.close();
        }
    }
}
